package future;

import org.junit.Test;

import java.util.concurrent.*;

/**
 * CompletableFuture API 测试
 *
 * @author codefish
 * @version 1.0
 * @date 2022/07/29 上午 11:19
 */
public class CompletableFutureApiTest {

    // ----------------------------------------------------------------------
    // --                                  获得结果和触发计算                 --
    // ----------------------------------------------------------------------

    /**
     * get()获取异步结果，需要处理ExecutionException和InterruptedException
     *
     * @throws ExecutionException   ex1
     * @throws InterruptedException ex2
     */
    @Test
    public void testGet() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "abc";
        });
        //get()获取异步计算结果
        System.out.println(future.get());
    }

    /**
     * get(long timeout, TimeUnit unit)在限定时间内获取异步结果，超时则抛出TimeoutException
     *
     * @throws ExecutionException   ex1
     * @throws InterruptedException ex2
     * @throws TimeoutException     ex3
     */
    @Test
    public void testGetWithTimeout() throws ExecutionException, InterruptedException, TimeoutException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "abc";
        });
        //get()获取异步计算结果
        System.out.println(future.get(3, TimeUnit.SECONDS));
    }

    /**
     * join()获取异步计算结果,不需要显式处理异常，在运行时抛出异常
     */
    @Test
    public void testJoin() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "abc";
        });
        //join()获取异步计算结果
        System.out.println(future.join());
    }

    /**
     * getNow(T valueIfAbsent)方法获取异步结果，当异步任务没有计算完成时，返回入参valueIfAbsent的值。
     */
    @Test
    public void testGetNow() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "abc";
        });
        //getNow()获取异步结果
        System.out.println(future.getNow("fallback"));
    }

    /**
     * complete(T value)在异步任务计算未完成时结束该任务,并且将入参value的值做为计算结果，保证程序在调用get()方法时返回value的值。
     */
    @Test
    public void testComplete() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "abc";
        });

        try {
            // TimeUnit.MILLISECONDS.sleep(2000);
            TimeUnit.MILLISECONDS.sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("是否打断？" + future.complete("fallback"));
        System.out.println(future.join());
    }

    // ----------------------------------------------------------------------
    // --                         对计算结果进行处理                          --
    // ----------------------------------------------------------------------

    /**
     * thenApply(Function<? super T,? extends U> fn)方法定义后续任务执行步骤，当某一步抛出异常后后续步骤都不执行
     */
    @Test
    public void testThenApply() {
        CompletableFuture<Integer> future = CompletableFuture
                .supplyAsync(() -> {
                    try {
                        TimeUnit.MILLISECONDS.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    int res = ThreadLocalRandom.current().nextInt(20);
                    System.out.println(currentThreadTag() + "第一步结果——" + res);
                    return res;
                })
                .thenApply(x -> {
                    int res = 2 * x;
                    try {
                        TimeUnit.MILLISECONDS.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(currentThreadTag() + "第二步结果——" + res);
                    return res;
                })
                .thenApply(x -> {
                    int res = x + 5;
                    //  res = res / 0;
                    try {
                        TimeUnit.MILLISECONDS.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(currentThreadTag() + "第三步结果——" + res);
                    return res;
                })
                .whenComplete((res, ex) -> {
                    if (ex == null) {
                        System.out.println(currentThreadTag() + "最终运行结果——" + res);
                    }
                })
                .exceptionally(ex -> {
                    ex.printStackTrace();
                    System.out.println("运算异常：" + ex.getMessage());
                    return null;
                });

        System.out.println(currentThreadTag() + "异步运算结果为：" + future.join());
    }

    /**
     * handle(BiFunction<? super T, Throwable, ? extends U> fn) 方法在某个步骤抛出异常后后续步骤仍然会执行,接收两个参数，
     * 第一个是上一步骤的计算结果(出现异常时为null)，第二个是上一步抛出的异常对象，为null则表示上一步未抛出异常
     */
    @Test
    public void testHandle() {
        CompletableFuture<Integer> future = CompletableFuture
                .supplyAsync(() -> {
                    try {
                        TimeUnit.MILLISECONDS.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    int res = ThreadLocalRandom.current().nextInt(20);
                    System.out.println(currentThreadTag() + "第一步结果——" + res);
                    return res;
                })
                .handle((x, ex) -> {
                    if (ex != null) {
                        int res = 2 * x;
                        res = res / 0;
                        try {
                            TimeUnit.MILLISECONDS.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println(currentThreadTag() + "第二步结果——" + res);
                        return res;
                    } else {
                        System.out.println(currentThreadTag() + "第二步异常——" + ex.getMessage());
                        return -1;
                    }

                })
                .handle((x, ex) -> {
                    if (ex == null) {
                        System.out.println(ex.getMessage());
                        System.out.println(x);
                        int res = x + 5;
                        try {
                            TimeUnit.MILLISECONDS.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println(currentThreadTag() + "第三步结果——" + res);
                        return res;
                    } else {
                        System.out.println(currentThreadTag() + "第三步异常——" + ex.getMessage());
                        return -1;
                    }
                })
                .whenComplete((res, ex) -> {
                    if (ex == null) {
                        System.out.println(currentThreadTag() + "最终运行结果——" + res);
                    }
                }).exceptionally(ex -> {
                    ex.printStackTrace();
                    System.out.println("运算异常：" + ex.getMessage());
                    return -2;
                });
        System.out.println(currentThreadTag() + "异步运算结果为：" + future.join());
    }

    // ----------------------------------------------------------------------
    // --                                  对计算结果进行消费                 --
    // ----------------------------------------------------------------------

    //除thenApply()外，thenRun()和thenAccept()都可以对计算结果进行消费,相比于thenApply()，它们没有返回值(返回的是CompletableFuture<Void>)对象

    @Test
    public void testThenApply2() {
        System.out.println(CompletableFuture.supplyAsync(() -> 4).thenApply(x -> "hahaha" + x).join());
    }

    @Test
    public void testThenRun() {
        System.out.println(CompletableFuture.supplyAsync(() -> 4).thenRun(() -> {
            System.out.println("hahaha");
        }).join());
    }

    @Test
    public void testThenAccept() {
        System.out.println(CompletableFuture.supplyAsync(() -> 4).thenAccept(System.out::println).join());
    }

    /**
     * thenRunAsync(),执行该方法时如果不指定线程池，则适用默认的ForkJoinPool,而thenRun()则沿用前面适用的线程池。其它thenXXXAsync()方法同理
     */
    @Test
    public void testThenRunAsync() {
        ExecutorService threadPool = Executors.newFixedThreadPool(5);

        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(currentThreadTag() + "2s");
        }, threadPool).thenRun(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(currentThreadTag() + "3s");
        }).thenRunAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(currentThreadTag() + "5s");
        }).thenRun(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(currentThreadTag() + "5s");
        });
        future.join();

    }

    // ----------------------------------------------------------------------
    // --                                  对计算速度进行选用                 --
    // ----------------------------------------------------------------------

    /**
     * applyToEither()，计算时间短的异步任务的结果被返回
     */
    @Test
    public void testApplyToEither() {
        CompletableFuture<String> playerA = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "playerA";
        });
        CompletableFuture<String> playerB = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "playerB";
        });
        CompletableFuture<Object> resFuture = playerA.applyToEither(playerB, res -> {
            return res + " is win";
        });

        System.out.println(resFuture.join());
    }

    /**
     * thenCombine(),计算结果合并
     */
    @Test
    public void testThenCombine() {
        CompletableFuture<Integer> process1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 50;
        });
        CompletableFuture<Integer> process2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 40;
        });

        CompletableFuture<Integer> future = process1.thenCombine(process2, (x1, x2) -> x1 + x2);
        System.out.println(future.join());
    }

    @Test
    public void test() {
        CompletableFuture<Integer> aaa = CompletableFuture.supplyAsync(() -> {
            System.out.println(currentThreadTag());
            try {
                TimeUnit.MILLISECONDS.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 4;
        }).thenCompose(x -> {
                    System.out.println(currentThreadTag());
                    try {
                        TimeUnit.MILLISECONDS.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return CompletableFuture.supplyAsync(() -> x + 5);
                }

        );
        System.out.println(aaa.join());

    }


    private String currentThreadTag() {
        return "[" + Thread.currentThread().getName() + "]：";
    }


}
